package com.eshipenet.shipnet.edgeboxextbuilder
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema
import play.api.libs.json.{Format, JsObject, Json}

case class EdgeBoxExtActionPayload(op: String, before: Option[JsObject], after: Option[JsObject], ts_ms: Long)
case class EdgeBoxExtActionKafka(schema: JsObject = Json.obj(), payload: EdgeBoxExtActionPayload)

object EdgeBoxExtActionPayload {
  implicit val format: Format[EdgeBoxExtActionPayload] = Json.format[EdgeBoxExtActionPayload]
}
object EdgeBoxExtActionKafka {
  implicit val format: Format[EdgeBoxExtActionKafka] = Json.format[EdgeBoxExtActionKafka]
}

class EdgeBoxExtActionKafkaSerializer(topic: String) extends KeyedSerializationSchema[EdgeBoxExtAction] {
  override def serializeKey(element: EdgeBoxExtAction): Array[Byte] = element.id.getBytes()

  override def serializeValue(element: EdgeBoxExtAction): Array[Byte] = {
    val ext = element match {
      case EdgeBoxExtInsert(_, action, model) =>
        val after = action.after.data + ("model" -> Json.toJson(model))
        EdgeBoxExtActionKafka(payload =
          EdgeBoxExtActionPayload(op = "i", before = None, after = Some(after), ts_ms = action.raw.timestamp.getMillis))
      case EdgeBoxExtUpdate(_, action, model) =>
        val before = action.before.data + ("model" -> Json.toJson(model))
        val after  = action.after.data + ("model"  -> Json.toJson(model))
        EdgeBoxExtActionKafka(
          payload = EdgeBoxExtActionPayload(op = "u",
                                            before = Some(before),
                                            after = Some(after),
                                            ts_ms = action.raw.timestamp.getMillis))
      case EdgeBoxExtDelete(_, action) =>
        val before = action.before.data
        EdgeBoxExtActionKafka(
          payload = EdgeBoxExtActionPayload(op = "d",
                                            before = Some(before),
                                            after = None,
                                            ts_ms = action.raw.timestamp.getMillis)
        )
      case EdgeBoxExtModelUpdate(_, edgeBox, action) =>
        val before = None
        val after  = edgeBox.data + ("model" -> Json.toJson(action))
        EdgeBoxExtActionKafka(
          payload = EdgeBoxExtActionPayload(op = "u", before = before, after = Some(after), ts_ms = 0)
        )
    }
    Json.toJson(ext).toString().getBytes()
  }

  override def getTargetTopic(element: EdgeBoxExtAction): String = topic

}
